package gotask

import (
	"fmt"
	log "github.com/sirupsen/logrus"
	"time"
	redis "github.com/go-redis/redis/v7"
)

var (
	_ Broker = &RedisBroker{}

	// rc: RedisClient
	rc *redis.Client
)

func genTaskName(taskID string) string {
	return "gt:task:" + taskID
}

func genQueueName(queueName string) string {
	return "gt:queue:" + queueName
}

type RedisBroker struct {
	TaskTTL int
}

type RedisBrokerOption func(rb *RedisBroker)

func WithRedisTaskTTL(ttl int) RedisBrokerOption {
	return func(rb *RedisBroker) {
		rb.TaskTTL = ttl
	}
}

func UseRedisBroker(redisURL string, brokerOptions ...RedisBrokerOption)  error {
	options, err := redis.ParseURL(redisURL)
	if err != nil {
		log.WithError(err).Panicf("failed to parse redis URL %s", redisURL)
	}

	rc = redis.NewClient(options)
	pong, err := rc.Ping().Result()
	if err != nil {
		log.WithError(err).Errorf("连接redis报错，请检查.")
		return err
	} else {
		log.WithField("Pong",pong).Info("connect redis sucessful")
	}

	rb := &RedisBroker{}
	for _, o := range brokerOptions {
		o(rb)
	}

	broker = rb
	return nil
}

func (r *RedisBroker) Acquire(queueName string) (*Task, error) {
	task := Task{}
	vs, err := rc.BRPop(time.Duration(0), genQueueName(queueName)).Result()
	if err != nil {
		log.WithError(err).Error("failed to get task from redis.")
		return nil, err // never executed
	}
	v := []byte(vs[1])

	if err := json.Unmarshal(v, &task); err != nil {
		log.WithError(err).Error("failed to Unmarshal task from redis")
		return nil, err // never executed
	}
	log.Infof("aquired a task[%s] from queue[%s]", task.ID, genQueueName(queueName))
	task.Status = ACQUIRED
	task.ResultLog = fmt.Sprintf("%s\n%s\n",task.ResultLog, "acquired")
	if err := r.Update(&task);err != nil {
		log.WithError(err).Error("update task info with marshal error")
		return nil, err
	}
	log.Infof("update task[%s] status acquired", task.ID)
	return &task, nil
}

func (r *RedisBroker) Ack(task *Task) bool {
	// redis doesn't support ACK
	return true
}

func (r *RedisBroker) Update(task *Task) (error) {
	task.UpdatedAt = time.Now()
	taskBytes, err := json.Marshal(task)
	if err != nil {
		log.WithError(err).Errorf("failed to enquue task %s.", task.ID)
		return  err// never executed here
	}
	log.Infof("update task[%s]", task.ID)
	rc.Set(genTaskName(task.ID), taskBytes, time.Duration(r.TaskTTL)*time.Second)
	return nil
}

func (r *RedisBroker) Enqueue(task *Task) string {
	task.Status = ENQUEUED
	taskBytes, err := json.Marshal(task)
	if err != nil {
		log.WithError(err).Error("failed to Marshal task %s", task.ID)
		return "" // never executed here
	}

	rc.Set(genTaskName(task.ID), taskBytes, time.Duration(r.TaskTTL)*time.Second)
	log.Infof("set task[%s]", genTaskName(task.ID))
	rc.LPush(genQueueName(task.QueueName), taskBytes)
	log.Infof("push task into queue[%s]",genQueueName(task.QueueName) )
	return task.ID
}

func (r *RedisBroker) QueueLen(queueName string) int64 {
	l, _ := rc.LLen(genQueueName(queueName)).Result()
	return l
}
